package com.xzx.mr.io;

import com.xzx.mr.kv.AnalysisKey;
import com.xzx.mr.kv.AnalysisValue;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;
import java.sql.*;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/8/10 19:23
 */
@Slf4j
public class MysqlWriter extends RecordWriter<AnalysisKey, AnalysisValue> {
    private Connection connection;

    public MysqlWriter() {
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "Admin_111");
        } catch (Exception e) {
            log.error("get connection error", e);
        }
    }

    @Override
    public void write(AnalysisKey key, AnalysisValue value) throws IOException, InterruptedException {
        log.info("====write: key:{}--> value:{}", key, value);
        try {
            PreparedStatement preparedStatement = connection.prepareStatement("insert into test.dianxin_call(phone, dt, cnt, duration) values(?, ?,?,?)");
            preparedStatement.setString(1, key.getPhone());
            preparedStatement.setString(2, key.getDate());
            preparedStatement.setInt(3, value.getSumCnt());
            preparedStatement.setLong(4, value.getSumDuration());
            preparedStatement.executeUpdate();
        } catch (SQLException e) {
            log.error("write error", e);
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                log.error("close error", e);
            }
        }
    }
}
